/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.ulss.indexer.metric;

import cn.ac.iie.ulss.indexer.runenvs.GlobalParas;
import cn.ac.iie.ulss.indexer.mqproducer.RocketMQProducer;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.SendStatus;
import com.alibaba.rocketmq.common.message.Message;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.logging.Level;
import org.apache.avro.Protocol;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.EncoderFactory;
import org.apache.log4j.Logger;

/**
 *
 * 
 */
public class DataIndexerMetricReportor {

    public static Logger log = Logger.getLogger(DataIndexerMetricReportor.class.getName());
    public static Schema metricSchema = null;
    public static boolean metricEnable = true;
    public static volatile boolean stopped;
    static ArrayBlockingQueue<MetricInfo> metricMsgCache = null;
    public static String metricMQ = null;
    public static DefaultMQProducer producer;

    public static void init() throws Exception {
        metricSchema = null;
        try {
            Protocol protocol = Protocol.parse(new File("data_volume_stat.json"));
            metricSchema = protocol.getType("data_volume_stat");
        } catch (Exception ex) {
            throw ex;
        }
        metricEnable = GlobalParas.isSendMetric;
        metricMQ = GlobalParas.staticsMq;
        metricMsgCache = new ArrayBlockingQueue<MetricInfo>((GlobalParas.metricBuffersize));
        stopped = false;
        DataIndexerMetricReportor.producer = RocketMQProducer.producer;

        for (int i = 0; i < GlobalParas.metricMsgDispatchWorkerNum; i++) {
            DispatcherWorker dispatcherWorker = new DispatcherWorker();
            dispatcherWorker.start();
        }
        log.info(metricEnable + " " + metricMQ + " " + stopped + " " + DataIndexerMetricReportor.producer);
    }

    public static void shutdown() {
        stopped = true;
    }

    static class MetricInfo {

        String docSchemaName;
        String region;
        long volume;
        String volume_type;
        String direction;
        String remark;
        String stat_source;

        public MetricInfo(String pDocSchemaName, String pRegion, long pVolume, String volume_type, String direction, String remark, String stat_source) {
            this.docSchemaName = pDocSchemaName;
            this.region = pRegion;
            this.volume = pVolume;
            this.volume_type = volume_type;
            this.direction = direction;
            this.remark = remark;
            this.stat_source = stat_source;
        }
    }

    public static void sendMetric(String objID, String pDocSchemaName, String pRegion, long pVolume, String volume_type, String direction, String remark, String stat_source) {
        if (!metricEnable) {
            return;
        }
        MetricInfo metricInfo = new MetricInfo(pDocSchemaName, pRegion, pVolume, volume_type, direction, remark, stat_source);

        int tryTimes = 0;
        while (true) {
            if (metricMsgCache.offer(metricInfo)) {
                break;
            } else {
                if ((++tryTimes) >= 20) {
                    log.warn("put msg into metric msg cache failed one time ");
                }
                try {
                    Thread.sleep(5);
                } catch (InterruptedException ex) {
                }
            }
        }
    }

    static class DispatcherWorker extends Thread {         //

        @Override
        public void run() {
            log.info("dispatcher worker for metric " + Thread.currentThread().getId() + " starts to run ");
            MetricInfo metricInfo = null;
            Message msg = null;
            while (true) {
                if (stopped) {
                    int msgCacheSize = metricMsgCache.size();
                    if (msgCacheSize <= 0) {
                        producer.shutdown();
                        break;
                    } else {
                        log.info("waiting " + msgCacheSize + " msg of " + metricMsgCache + " to be cleared...");
                    }
                }
                try {
                    metricInfo = metricMsgCache.poll();
                    if (metricInfo == null) {
                        try {
                            Thread.sleep(100);
                        } catch (Exception ex) {
                        }
                        continue;
                    }
                    GenericRecord metricRecord = new GenericData.Record(metricSchema);
                    metricRecord.put("doc_schema_name", metricInfo.docSchemaName);
                    metricRecord.put("region", metricInfo.region);
                    metricRecord.put("volume", metricInfo.volume);
                    metricRecord.put("volume_type", metricInfo.volume_type);
                    metricRecord.put("direction", metricInfo.direction);
                    metricRecord.put("remark", metricInfo.remark);
                    metricRecord.put("stat_timestamp", new Date().getTime());
                    metricRecord.put("stat_source", metricInfo.stat_source);
                    metricRecord.put("src_ip", GlobalParas.ip);

                    DatumWriter<GenericRecord> metricWriter = new GenericDatumWriter<GenericRecord>(metricSchema);
                    ByteArrayOutputStream metricsos = new ByteArrayOutputStream();
                    BinaryEncoder metricen = new EncoderFactory().binaryEncoder(metricsos, null);
                    try {
                        metricWriter.write(metricRecord, metricen);
                        metricen.flush();
                        msg = new Message(metricMQ, metricsos.toByteArray());
                    } catch (Exception ex) {
                        log.warn("do metric obj serilization for [" + metricRecord + "] unsuccessfully for " + ex.getMessage(), ex);
                        msg = null;
                    } finally {
                        metricRecord = null;
                    }
                    if (msg != null) {
                        SendResult sr = null;
                        long startTime = System.nanoTime();
                        while (sr == null || sr.getSendStatus() != SendStatus.SEND_OK) {
                            sr = DataIndexerMetricReportor.producer.send(msg);
                            if (sr == null || sr.getSendStatus() == SendStatus.FLUSH_DISK_TIMEOUT) {
                                log.warn("send message fail ,will sleep and retry,information:" + producer.getClientIP() + " " + producer.getProducerGroup());
                                try {
                                    Thread.sleep(200);
                                } catch (Exception e) {
                                }
                                continue;
                            } else {
                                long endTime = System.nanoTime();
                                log.debug("send metric message to " + sr.getMessageQueue().getBrokerName() + ":" + sr.getQueueOffset() + " ok:" + (endTime - startTime) / (1000 * 1000) + " status:" + sr.getSendStatus());
                                break;
                            }
                        }
                        msg = null;
                    } else {
                        try {
                            Thread.sleep(100);
                        } catch (Exception ex) {
                        }
                    }
                } catch (Exception ex) {
                    log.warn("error happened when sending metric messge " + ex.getMessage(), ex);
                }

                if (stopped) {
                    int msgCacheSize = metricMsgCache.size();
                    if (msgCacheSize <= 0) {
                        producer.shutdown();
                        break;
                    } else {
                        log.info("waiting " + msgCacheSize + " msg of " + metricMsgCache + " to be cleared...");
                    }
                }
            }
        }
    }
}
